package com.jushu.storbox.task.tqueue;

import java.util.Collection;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ShareDownloadPriorityTaskScheduler <T extends ShareDownloadPriorityTask>{
	
	private final PriorityBlockingQueue<T> taskQueue;
	private final ExecutorService service;
	private final ExecutorService runningTasks;
	private volatile boolean stop = false;
	
	public ShareDownloadPriorityTaskScheduler(int initialCapacity) {
		taskQueue = new PriorityBlockingQueue<T>(initialCapacity, new Comparator<T>() {

			@Override
			public int compare(T arg0, T arg1) {
				// TODO Auto-generated method stub
				if(null == arg0 && null == arg1) return 0;
				if (null == arg0 || null == arg1) {
					if (null == arg0) return -1;
					else return 1;
				}
				if (arg0.hashCode() == arg1.hashCode())
					return 0;
//				if (arg0.getId().equalsIgnoreCase(arg1.getId()))
//					return 0;
				int p1 = arg1.getPriority();
				int p2 = arg0.getPriority();
				
				if(p1 != p2)
					return arg1.getPriority() - arg0.getPriority();
				else
					return (int)(arg1.getSequenceNumber() - arg0.getSequenceNumber());
			}
			
		});
		service = Executors.newSingleThreadExecutor();
//		runningTasks = Executors.newFixedThreadPool(initialCapacity);
		runningTasks = new ThreadPoolExecutor(initialCapacity, initialCapacity, 0L,
				TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1000));
	}
	
	public Set<T> getTaskQueue() {
		return new HashSet(taskQueue);
	}
	
	public void runTask(T task) {
		synchronized(taskQueue) {
			taskQueue.add(task);
			taskQueue.notifyAll();
		}
	}
	
	public synchronized void runTask(Collection <T> c) {
		synchronized(taskQueue) {
			taskQueue.addAll(c);
			taskQueue.notifyAll();
		}
	}
	
	public boolean removeTask(T task) {
		synchronized(taskQueue) {
			return taskQueue.remove(task);
		}
	}
	
	public void stop() {
		stop = true;
		synchronized(taskQueue) {
			taskQueue.notifyAll();
		}
		runningTasks.shutdownNow();
		service.shutdownNow();
	}
	
	public int size() {
		return taskQueue.size();
	}
	
	public boolean contains(T task){
		return taskQueue.contains(task);
	}
	
	public synchronized void start() {
		service.execute(new Runnable() {
			@Override
			public void run() {
				// TODO Auto-generated method stub
				while(!stop) {
					try {
						synchronized(taskQueue) {
							taskQueue.wait();
						}
						if(stop) break;
						while(!taskQueue.isEmpty() && !Thread.interrupted() && !stop) {
							T task = taskQueue.poll();
							try {
								runningTasks.execute(task);
							} catch(Exception ex) {
								ex.printStackTrace();
							}
						}
					} catch(Exception e) {
					}
				}
			}
		});
	}

	
	
	
}
